package com.nami.dispatcher.controller.middleware.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.google.protobuf.InvalidProtocolBufferException;
import com.nami.dispatcher.facade.grpmsg.ForwardGrpMsgFacade;
import com.nami.dispatcher.facade.grpmsg.request.command.GrpDispatchCmd;
import com.nami.gate.external.basic.bo.ConnectedUserBO;
import com.nami.gate.external.basic.constants.DispatcherTopicConst;

import com.nami.im.network.model.message.v2.Message;
import com.nami.transport.proto.codec.MqTransMsg;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = DispatcherTopicConst.GCHAT_TOPIC,
    consumerGroup = DispatcherTopicConst.GCHAT_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class GrpMsgConsumer implements RocketMQListener<MessageExt> {

    private final ForwardGrpMsgFacade sendGrpMsgCmdExe;

    @Override
    public void onMessage(MessageExt msg) {
        TimeInterval timer = DateUtil.timer();
        GrpDispatchCmd cmd = parseGrpMsgCmd(msg.getBody());
        sendGrpMsgCmdExe.execute(cmd);
        Message message = cmd.getMessage();
        if (timer.interval() > 200) {
            log.warn("群聊消息出现处理慢，msgId={}，clientMsgId={}， use:{}ms",
                message.getMsgId(), message.getClientUid(), timer.interval());
        }
    }

    private GrpDispatchCmd parseGrpMsgCmd(byte[] msg) {
        try {
            MqTransMsg.GrpLogic2DispatchBody request = MqTransMsg.GrpLogic2DispatchBody.parseFrom(msg);
            Message message  = JsonUtils.parseObject(request.getMessage().toStringUtf8(), Message.class);
            GrpDispatchCmd command = new GrpDispatchCmd();
            command.setQos(request.getQos());
            command.setMessage(message);

            command.setSuccess(request.getSuccess());
            command.setErrCode(request.getErrCode());
            command.setErrMessage(request.getErrMessage());


            MqTransMsg.TransUser sender = request.getSender();
            ConnectedUserBO senderUserCO = new ConnectedUserBO();
            senderUserCO.setAppKey(sender.getAppKey());
            senderUserCO.setUserId(sender.getUserId());
            senderUserCO.setPlatform(sender.getPlatform());
            senderUserCO.setConnectId(sender.getConnectId());
            senderUserCO.setInstanceId(sender.getInstanceId());
            command.setSender(senderUserCO);

            command.setMemberUserIds(request.getReceiverIdsList());


            return command;
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        }

    }


}
